/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package com.imsx.controller.util;

import com.basho.riak.client.api.RiakClient;
import com.basho.riak.client.api.commands.kv.DeleteValue;
import com.basho.riak.client.api.commands.kv.FetchValue;
import com.basho.riak.client.api.commands.kv.StoreValue;
import com.basho.riak.client.core.RiakCluster;
import com.basho.riak.client.core.RiakNode;
import com.basho.riak.client.core.query.Location;
import com.basho.riak.client.core.query.Namespace;
import com.basho.riak.client.core.query.RiakObject;
import com.basho.riak.client.core.util.BinaryValue;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ConexionRiak {

    RiakCluster cluster;
    RiakClient client;

    private static RiakCluster setUpCluster() throws UnknownHostException {
        // This example will use only one node listening on localhost:10017
        RiakNode node1 = new RiakNode.Builder()
                .withRemoteAddress("192.168.1.101")
                .withRemotePort(8087)
                .build();
    

        RiakNode node2 = new RiakNode.Builder()
                .withRemoteAddress("192.168.1.102")
                .withRemotePort(8087)
                .build();
        RiakNode node3 = new RiakNode.Builder()
                .withRemoteAddress("192.168.1.103")
                .withRemotePort(8087)
                .build();

        // This cluster object takes our one node as an argument
        List<RiakNode> listaRiak = new ArrayList<RiakNode>();
        listaRiak.add(node1);
        listaRiak.add(node2);
        listaRiak.add(node3);

        RiakCluster cluster = new RiakCluster.Builder(listaRiak).build();

        // The cluster must be started to work, otherwise you will see errors
        cluster.start();

        return cluster;
    }

    public RiakClient conectar() {
        try {
            RiakObject quoteObject = new RiakObject()
                    // We tell Riak that we're storing plaintext, not JSON, HTML, etc.
                    .setContentType("text/plain")
                    // Objects are ultimately stored as binaries
                    .setValue(BinaryValue.create("You're dangerous, Maverick"));
           // System.out.println("Basic object created");

            // In the new Java client, instead of buckets you interact with Namespace
            // objects, which consist of a bucket AND a bucket type; if you don't
            // supply a bucket type, "default" is used; the Namespace below will set
            // only a bucket, without supplying a bucket type
            Namespace quotesBucket = new Namespace("quotes");

            // With our Namespace object in hand, we can create a Location object,
            // which allows us to pass in a key as well
            Location quoteObjectLocation = new Location(quotesBucket, "Iceman");
            //System.out.println("Location object created for quote object");

            // With our RiakObject in hand, we can create a StoreValue operation
            StoreValue storeOp = new StoreValue.Builder(quoteObject)
                    .withLocation(quoteObjectLocation)
                    .build();
           // System.out.println("StoreValue operation created");

            // And now we can use our setUpCluster() function to create a cluster
            // object which we can then use to create a client object and then
            // execute our storage operation
            cluster = setUpCluster();
            client = new RiakClient(cluster);
            //System.out.println("Client object successfully created");

            StoreValue.Response storeOpResp = client.execute(storeOp);
            //System.out.println("Object storage operation successfully completed");

            // Now we can verify that the object has been stored properly by
            // creating and executing a FetchValue operation
            FetchValue fetchOp = new FetchValue.Builder(quoteObjectLocation)
                    .build();
            RiakObject fetchedObject = client.execute(fetchOp).getValue(RiakObject.class);
            assert (fetchedObject.getValue().equals(quoteObject.getValue()));
           // System.out.println("Success! The object we created and the object we fetched have the same value");

            // And we'll delete the object
            DeleteValue deleteOp = new DeleteValue.Builder(quoteObjectLocation)
                    .build();
            client.execute(deleteOp);
            //System.out.println("Quote object successfully deleted");
            return client;
        } catch (ExecutionException ex) {
            Logger.getLogger(ConexionRiak.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InterruptedException ex) {
            Logger.getLogger(ConexionRiak.class.getName()).log(Level.SEVERE, null, ex);
        } catch (UnknownHostException ex) {
            Logger.getLogger(ConexionRiak.class.getName()).log(Level.SEVERE, null, ex);
        }
        return null;
    }

    public void cerrar() {
        cluster.shutdown();

    }
}
